package protoactor

import (
	"context"
	"fmt"
	"log"
	"os"
	"sync"
	"time"

	"github.com/asynkron/protoactor-go/actor"
	"github.com/asynkron/protoactor-go/cluster"
	"github.com/asynkron/protoactor-go/cluster/automanaged"
	"github.com/asynkron/protoactor-go/remote"
	"github.com/google/uuid"
	clusterapi "github.com/louloulin/dataflare/pkg/cluster"
)

// Manager implements the cluster.Manager interface using protoactor-go cluster
type Manager struct {
	// config is the cluster configuration
	config *clusterapi.Config
	// actorSystem is the actor system
	actorSystem *actor.ActorSystem
	// remote is the remote configuration
	remote *remote.Remote
	// cluster is the protoactor-go cluster
	cluster *cluster.Cluster
	// provider is the cluster provider
	provider *automanaged.AutoManagedClusterProvider
	// localNode is information about the local node
	localNode *clusterapi.NodeInfo
	// nodes is a map of node IDs to node information
	nodes map[string]*clusterapi.NodeInfo
	// eventChannels is a map of event types to channels
	eventChannels map[string][]chan clusterapi.Event
	// isLeader indicates if this node is the leader
	isLeader bool
	// leaderID is the ID of the current leader
	leaderID string
	// mu is a mutex for protecting shared state
	mu sync.RWMutex
	// logger is the logger for the cluster manager
	logger *log.Logger
	// ctx is the context for the cluster manager
	ctx context.Context
	// cancel is the cancel function for the context
	cancel context.CancelFunc
}

// NewManager creates a new protoactor-go cluster manager
func NewManager(config *clusterapi.Config) (clusterapi.Manager, error) {
	// Validate configuration
	if err := config.Validate(); err != nil {
		return nil, fmt.Errorf("invalid cluster configuration: %w", err)
	}

	// Generate node ID if not provided
	if config.NodeID == "" {
		config.NodeID = uuid.New().String()
	}

	// Create context
	ctx, cancel := context.WithCancel(context.Background())

	// Create actor system
	actorSystem := actor.NewActorSystem()

	// Create local node info
	localNode := &clusterapi.NodeInfo{
		ID:        config.NodeID,
		Address:   config.BindAddress,
		Role:      config.Role,
		Status:    "starting",
		Tags:      config.Tags,
		Metadata:  config.Metadata,
		StartTime: time.Now(),
		LastSeen:  time.Now(),
	}

	// Create cluster manager
	manager := &Manager{
		config:        config,
		actorSystem:   actorSystem,
		localNode:     localNode,
		nodes:         make(map[string]*clusterapi.NodeInfo),
		eventChannels: make(map[string][]chan clusterapi.Event),
		logger:        log.New(os.Stderr, "[ProtoActorCluster] ", log.LstdFlags),
		ctx:           ctx,
		cancel:        cancel,
	}

	// Add local node to nodes map
	manager.nodes[localNode.ID] = localNode

	return manager, nil
}

// Start starts the cluster manager
func (m *Manager) Start(ctx context.Context) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.logger.Printf("Starting protoactor-go cluster manager for node %s", m.config.NodeID)

	// Parse bind address
	host, port, err := parseAddress(m.config.BindAddress)
	if err != nil {
		return fmt.Errorf("invalid bind address: %w", err)
	}

	// Configure remote
	remoteConfig := remote.Configure(host, port)
	m.remote = remote.NewRemote(m.actorSystem, remoteConfig)
	err = m.remote.Start()
	if err != nil {
		return fmt.Errorf("failed to start remote: %w", err)
	}

	// Configure cluster provider
	providerConfig := automanaged.NewWithConfig(
		m.config.HeartbeatInterval,
		m.config.HeartbeatTimeout,
		m.config.AdvertiseAddress,
	)
	
	// Add seed nodes if provided
	if len(m.config.SeedAddresses) > 0 {
		for _, seed := range m.config.SeedAddresses {
			providerConfig.AddMember(m.config.AdvertiseAddress, seed)
		}
	}
	
	m.provider = providerConfig

	// Configure cluster
	clusterName := "dataflare-cluster"
	clusterConfig := cluster.Configure(
		clusterName,
		m.remote,
		cluster.NewPartitionIdentityLookup(),
	)

	// Create and start cluster
	m.cluster = cluster.New(m.actorSystem, clusterConfig, m.provider)
	err = m.cluster.Start()
	if err != nil {
		m.remote.Shutdown(true)
		return fmt.Errorf("failed to start cluster: %w", err)
	}

	// Update local node status
	m.localNode.Status = "running"

	// Start monitoring
	go m.monitorCluster()

	// Broadcast node join event
	m.broadcastEvent(clusterapi.Event{
		Type:      clusterapi.EventTypeNodeJoin,
		NodeID:    m.config.NodeID,
		Node:      m.localNode,
		Timestamp: time.Now(),
	})

	m.logger.Printf("Protoactor-go cluster manager started successfully")
	return nil
}

// Stop stops the cluster manager
func (m *Manager) Stop(ctx context.Context) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.logger.Printf("Stopping protoactor-go cluster manager for node %s", m.config.NodeID)

	// Update local node status
	m.localNode.Status = "stopping"

	// Broadcast node leave event
	m.broadcastEvent(clusterapi.Event{
		Type:      clusterapi.EventTypeNodeLeave,
		NodeID:    m.config.NodeID,
		Node:      m.localNode,
		Timestamp: time.Now(),
	})

	// Stop cluster
	if m.cluster != nil {
		err := m.cluster.Shutdown(true)
		if err != nil {
			m.logger.Printf("Error shutting down cluster: %v", err)
		}
	}

	// Stop remote
	if m.remote != nil {
		m.remote.Shutdown(true)
	}

	// Cancel context
	m.cancel()

	// Close all event channels
	for eventType, channels := range m.eventChannels {
		for _, ch := range channels {
			close(ch)
		}
		delete(m.eventChannels, eventType)
	}

	m.logger.Printf("Protoactor-go cluster manager stopped successfully")
	return nil
}

// Join joins the cluster using the provided seed nodes
func (m *Manager) Join(ctx context.Context, addresses []string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.logger.Printf("Joining cluster with seed nodes: %v", addresses)

	// Join cluster
	for _, address := range addresses {
		err := m.provider.AddMember(m.config.AdvertiseAddress, address)
		if err != nil {
			m.logger.Printf("Failed to add member %s: %v", address, err)
			// Continue with other addresses
		}
	}

	return nil
}

// Leave leaves the cluster
func (m *Manager) Leave(ctx context.Context) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.logger.Printf("Leaving cluster")

	// Update local node status
	m.localNode.Status = "leaving"

	// Broadcast node leave event
	m.broadcastEvent(clusterapi.Event{
		Type:      clusterapi.EventTypeNodeLeave,
		NodeID:    m.config.NodeID,
		Node:      m.localNode,
		Timestamp: time.Now(),
	})

	return nil
}

// GetNodes returns information about all nodes in the cluster
func (m *Manager) GetNodes(ctx context.Context) ([]*clusterapi.NodeInfo, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	nodes := make([]*clusterapi.NodeInfo, 0, len(m.nodes))
	for _, node := range m.nodes {
		nodes = append(nodes, node)
	}
	return nodes, nil
}

// GetNode returns information about a specific node
func (m *Manager) GetNode(ctx context.Context, nodeID string) (*clusterapi.NodeInfo, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	node, ok := m.nodes[nodeID]
	if !ok {
		return nil, fmt.Errorf("node not found: %s", nodeID)
	}
	return node, nil
}

// GetLocalNode returns information about the local node
func (m *Manager) GetLocalNode(ctx context.Context) (*clusterapi.NodeInfo, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.localNode, nil
}

// IsLeader checks if the local node is the leader
func (m *Manager) IsLeader(ctx context.Context) (bool, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.isLeader, nil
}

// GetLeader returns the leader node
func (m *Manager) GetLeader(ctx context.Context) (*clusterapi.NodeInfo, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.leaderID == "" {
		return nil, fmt.Errorf("no leader elected")
	}

	leader, ok := m.nodes[m.leaderID]
	if !ok {
		return nil, fmt.Errorf("leader node not found: %s", m.leaderID)
	}

	return leader, nil
}

// Subscribe subscribes to cluster events
func (m *Manager) Subscribe(ctx context.Context, eventType string) (<-chan clusterapi.Event, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Create channel
	ch := make(chan clusterapi.Event, 100)

	// Add channel to event channels
	if _, ok := m.eventChannels[eventType]; !ok {
		m.eventChannels[eventType] = make([]chan clusterapi.Event, 0)
	}
	m.eventChannels[eventType] = append(m.eventChannels[eventType], ch)

	return ch, nil
}

// Unsubscribe unsubscribes from cluster events
func (m *Manager) Unsubscribe(ctx context.Context, eventType string, ch <-chan clusterapi.Event) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Find and remove channel
	channels, ok := m.eventChannels[eventType]
	if !ok {
		return nil
	}

	for i, c := range channels {
		if c == ch {
			// Remove channel
			m.eventChannels[eventType] = append(channels[:i], channels[i+1:]...)
			close(c)
			break
		}
	}

	return nil
}

// Broadcast broadcasts a message to all nodes in the cluster
func (m *Manager) Broadcast(ctx context.Context, message []byte) error {
	// Not directly supported by protoactor-go cluster
	// Would need to implement custom message broadcasting
	return fmt.Errorf("broadcast not implemented")
}

// Send sends a message to a specific node
func (m *Manager) Send(ctx context.Context, nodeID string, message []byte) error {
	// Not directly supported by protoactor-go cluster
	// Would need to implement custom message sending
	return fmt.Errorf("send not implemented")
}

// SetMetadata sets node metadata
func (m *Manager) SetMetadata(ctx context.Context, key string, value string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Update local node metadata
	if m.localNode.Metadata == nil {
		m.localNode.Metadata = make(map[string]string)
	}
	m.localNode.Metadata[key] = value

	// Broadcast node update event
	m.broadcastEvent(clusterapi.Event{
		Type:      clusterapi.EventTypeNodeUpdate,
		NodeID:    m.config.NodeID,
		Node:      m.localNode,
		Timestamp: time.Now(),
	})

	return nil
}

// GetMetadata gets node metadata
func (m *Manager) GetMetadata(ctx context.Context, nodeID string, key string) (string, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Get node
	node, ok := m.nodes[nodeID]
	if !ok {
		return "", fmt.Errorf("node not found: %s", nodeID)
	}

	// Get metadata
	if node.Metadata == nil {
		return "", fmt.Errorf("metadata not found: %s", key)
	}

	value, ok := node.Metadata[key]
	if !ok {
		return "", fmt.Errorf("metadata not found: %s", key)
	}

	return value, nil
}

// monitorCluster monitors the cluster for changes
func (m *Manager) monitorCluster() {
	ticker := time.NewTicker(m.config.HeartbeatInterval)
	defer ticker.Stop()

	for {
		select {
		case <-m.ctx.Done():
			return
		case <-ticker.C:
			m.updateClusterState()
		}
	}
}

// updateClusterState updates the cluster state
func (m *Manager) updateClusterState() {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Get cluster topology
	topology := m.cluster.MemberList.GetPartitionMembersWithIndex(0)

	// Update nodes map
	currentNodes := make(map[string]bool)
	for _, member := range topology {
		nodeID := member.Id
		currentNodes[nodeID] = true

		// Add new node or update existing node
		if node, ok := m.nodes[nodeID]; ok {
			// Update existing node
			node.LastSeen = time.Now()
			node.Status = "running"
			node.Address = member.Address
		} else {
			// Add new node
			newNode := &clusterapi.NodeInfo{
				ID:        nodeID,
				Address:   member.Address,
				Status:    "running",
				Tags:      make([]string, 0),
				Metadata:  make(map[string]string),
				StartTime: time.Now(),
				LastSeen:  time.Now(),
			}
			m.nodes[nodeID] = newNode

			// Broadcast node join event
			m.broadcastEvent(clusterapi.Event{
				Type:      clusterapi.EventTypeNodeJoin,
				NodeID:    nodeID,
				Node:      newNode,
				Timestamp: time.Now(),
			})
		}
	}

	// Remove nodes that are no longer in the cluster
	for nodeID, node := range m.nodes {
		if nodeID == m.config.NodeID {
			// Skip local node
			continue
		}

		if !currentNodes[nodeID] {
			// Node is no longer in the cluster
			node.Status = "left"

			// Broadcast node leave event
			m.broadcastEvent(clusterapi.Event{
				Type:      clusterapi.EventTypeNodeLeave,
				NodeID:    nodeID,
				Node:      node,
				Timestamp: time.Now(),
			})

			// Remove node from nodes map
			delete(m.nodes, nodeID)
		}
	}

	// Update leader status
	// In protoactor-go, there's no built-in leader election
	// This is a simple implementation where the first node in the sorted list is the leader
	if len(m.nodes) > 0 {
		// Get sorted node IDs
		nodeIDs := make([]string, 0, len(m.nodes))
		for nodeID := range m.nodes {
			nodeIDs = append(nodeIDs, nodeID)
		}

		// Sort node IDs
		// In a real implementation, you would use a proper leader election algorithm
		// For simplicity, we just use the first node in the sorted list
		if len(nodeIDs) > 0 {
			newLeaderID := nodeIDs[0]
			if newLeaderID != m.leaderID {
				oldLeaderID := m.leaderID
				m.leaderID = newLeaderID
				m.isLeader = (m.leaderID == m.config.NodeID)

				// Broadcast leader changed event
				m.broadcastEvent(clusterapi.Event{
					Type:      clusterapi.EventTypeLeaderChanged,
					NodeID:    m.leaderID,
					Node:      m.nodes[m.leaderID],
					Timestamp: time.Now(),
					Data: map[string]interface{}{
						"old_leader_id": oldLeaderID,
					},
				})
			}
		}
	}
}

// broadcastEvent broadcasts an event to all subscribers
func (m *Manager) broadcastEvent(event clusterapi.Event) {
	// Send to specific event type subscribers
	if channels, ok := m.eventChannels[event.Type]; ok {
		for _, ch := range channels {
			select {
			case ch <- event:
				// Event sent
			default:
				// Channel full, log warning
				m.logger.Printf("Warning: event channel full, dropping event: %s", event.Type)
			}
		}
	}

	// Send to all event subscribers
	if channels, ok := m.eventChannels["*"]; ok {
		for _, ch := range channels {
			select {
			case ch <- event:
				// Event sent
			default:
				// Channel full, log warning
				m.logger.Printf("Warning: event channel full, dropping event: %s", event.Type)
			}
		}
	}
}

// parseAddress parses an address string into host and port
func parseAddress(address string) (string, int, error) {
	// For simplicity, we assume the address is in the format "host:port"
	// In a real implementation, you would use net.SplitHostPort
	host := "127.0.0.1"
	port := 7946

	return host, port, nil
}

// init registers the protoactor-go cluster manager factory
func init() {
	clusterapi.RegisterManagerFactory("protoactor", func(config *clusterapi.Config) (clusterapi.Manager, error) {
		return NewManager(config)
	})
}
